package com.gubatian.jdk.analyse.register.client;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicStampedReference;

/**
 * @Description: 定时拉取服务注册表
 * @auther: gubatian
 * @date: 2021/09/14
 */
public class ClientCachedServiceRegistry {
    /**
     * 每次拉取注册表的时间间隔
     */
    private final static Long SERVICE_REGISTRY_FETCH_INTERVAL = 30 * 1000L;
    /***
     * 注册表
     */
//    private AtomicReference<Applications> applications = new AtomicReference<>(new Applications());
    private AtomicStampedReference<Applications> applications;
    /**
     * 增量拉取注册表后台线程
     */
    private FetchDeltaRegistryWorker fetchDeltaRegistryWorker;
    private RegisterClient registerClient;
    private HttpSender httpSender;
    /**
     * 代表了本地缓存服务注册表的一个版本号
     */
    private AtomicLong applicationsVersion = new AtomicLong(0L);

    public ClientCachedServiceRegistry(RegisterClient registerClient, HttpSender httpSender) {
        this.registerClient = registerClient;
        this.httpSender = httpSender;
        this.fetchDeltaRegistryWorker = new FetchDeltaRegistryWorker();
        this.applications = new AtomicStampedReference<>(new Applications(), 0);
    }

    public void initialize () {
        // 启动全量拉取注册表
        FetchFullRegistryWorker fetchFullRegistryWorker = new FetchFullRegistryWorker();
        fetchFullRegistryWorker.start();

        // 启动增量拉取注册表
        this.fetchDeltaRegistryWorker.setDaemon(true);
        this.fetchDeltaRegistryWorker.start();
    }
    /**
     * 全量拉取注册表的后台线程
     */
    private class FetchFullRegistryWorker extends Thread {
        @Override
        public void run() {

            fetchFullRegistry();
        }
    }

    /**
     * 拉取全量注册表
     * 这个操作要走网络，如果因为网路原因这个地方一直卡住，数据拉取不到
     * 此时的数据是一个旧的版本，里面只有30个服务实例
     * 全量拉取注册表的线程突然苏醒过来
     * 把这30个旧的服务实例写入注册表
     *
     * 一定要在网络发起请求之前拿到当时注册表的一个版本号
     * 接着发起网路请求，如果这时候有线程已经修改了注册表，版本号也会跟着修改
     * 必须是发起网路请求后，这个注册表的版本号没有被修改过，它才能去修改
     * 此时你就不要把你拉取到的旧的注册表写入本地缓存中
     */
    public void fetchFullRegistry() {
        long expectVersion = applicationsVersion.get();
        if (applicationsVersion.compareAndSet(expectVersion, expectVersion + 1)) {
            // 拉取全量注册表
            Applications fetchApplications = httpSender.fetchFullRegistry();
            Applications expectedApplications;
            int expectedStamp;
            do {
                expectedApplications = applications.getReference();
                expectedStamp = applications.getStamp();
            } while (!applications.compareAndSet(
                    expectedApplications,
                    fetchApplications,
                    expectedStamp,
                    expectedStamp + 1));
        }
    }
    /**
     * 增量拉取注册表的后台线程
     */
    private class FetchDeltaRegistryWorker extends Thread {
        @Override
        public void run() {
            while (registerClient.isRunning()) {
                try {
                    // 增量注册表
                    DeltaRegistry deltaRegistry = httpSender.fetchDeltaRegistry();

                    // 最近变更的增量注册表队列有两种
                    // 注册、删除
                    // 如果是新注册的服务实例，跟全量注册表比对一下是否有这个实例，如果没有合并到全量注册表里
                    // 如果是删除的服务实例，遍历注册表，如果存在该服务删除之

                    Map<String, Map<String, ServiceInstance>> registry =
                            applications.getReference().getRegistry();
                    // 这里需要做大量的缓冲中注册表的修改，所以需要对注册表进行加锁
                    synchronized (registry) {
                        mergeDelta(deltaRegistry.getRecentlyChangedQueue());
                    }

                    // 再检查一下，跟服务端的注册表的服务实例的数量相比，是否是一致的
                    // 封装一下增量注册表的对象，也就是拉取增量注册表的时候，一方面是返回那个数据
                    // 另外一方面，是要那个对应的register-server端的服务实例的数量
                    reconcileRegistry(deltaRegistry);
                    Thread.sleep(SERVICE_REGISTRY_FETCH_INTERVAL);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }

        private void mergeDelta(LinkedList<RecentlyChangedServiceInstance> deltaRegistry) {
            Map<String, Map<String, ServiceInstance>> registry =
                    applications.getReference().getRegistry();
            for(RecentlyChangedServiceInstance recentlyChangedItem : deltaRegistry) {
                // 服务名称
                String serviceName = recentlyChangedItem.serviceInstance.getServiceName();
                // 服务实例ID
                String serviceInstanceId = recentlyChangedItem.serviceInstance.getServiceInstanceId();
                //
                Map<String, ServiceInstance> serviceInstanceMap = registry.get(serviceName);
                // 如果是注册操作的话
                if(ServiceInstanceOperation.REGISTER.equals(
                        recentlyChangedItem.serviceInstanceOperation)) {
                    if(serviceInstanceMap == null) {
                        serviceInstanceMap = new HashMap<>();
                        registry.put(serviceName, serviceInstanceMap);
                    }

                    ServiceInstance serviceInstance = serviceInstanceMap.get(serviceInstanceId);
                    if(serviceInstance == null) {
                        serviceInstanceMap.put(serviceInstanceId, recentlyChangedItem.serviceInstance);
                    }
                }
                // 如果是删除操作的话
                else if(ServiceInstanceOperation.REMOVE.equals(
                        recentlyChangedItem.serviceInstanceOperation)) {
                    if(serviceInstanceMap != null) {
                        serviceInstanceMap.remove(serviceInstanceId);
                    }
                }
            }
        }
        /**
         * 校对调整注册表
         * @param deltaRegistry
         */
        private void reconcileRegistry(DeltaRegistry deltaRegistry) {
            Map<String, Map<String, ServiceInstance>> registry =
                    applications.getReference().getRegistry();

            Long serverSideTotalCount = deltaRegistry.getServiceInstanceTotalCount();

            Long clientSideTotalCount = 0L;
            for(Map<String, ServiceInstance> serviceInstanceMap : registry.values()) {
                clientSideTotalCount += serviceInstanceMap.size();
            }

            if(serverSideTotalCount != clientSideTotalCount) {
                // 重新拉取全量注册表进行纠正
                Applications fetchApplications = httpSender.fetchFullRegistry();
                Applications expectedApplications;
                int expectedStamp;
                do {
                    expectedApplications = applications.getReference();
                    expectedStamp = applications.getStamp();
                } while (!applications.compareAndSet(
                        expectedApplications,
                        fetchApplications,
                        expectedStamp,
                        expectedStamp + 1));
            }
        }

    }
    public void destroy() {
        this.fetchDeltaRegistryWorker.interrupt();
    }

    public Map<String, Map<String, ServiceInstance>> getRegistry() {
        return applications.getReference().getRegistry();
    }

    /**
     * 最近变更的服务实例
     */
    static class RecentlyChangedServiceInstance {
        /**
         * 服务实例
         */
        ServiceInstance serviceInstance;
        /**
         * 变更时间
         */
        long changedTimeStamp;
        /**
         * 服务变更操作
         */
        String serviceInstanceOperation;
        public RecentlyChangedServiceInstance(ServiceInstance serviceInstance, long changedTimeStamp, String serviceInstanceOperation) {
            this.serviceInstance = serviceInstance;
            this.changedTimeStamp = changedTimeStamp;
            this.serviceInstanceOperation = serviceInstanceOperation;
        }

        @Override
        public String toString() {
            return "RecentlyChangedServiceInstance{" +
                    "serviceInstance=" + serviceInstance +
                    ", changedTimeStamp=" + changedTimeStamp +
                    ", serviceInstanceOption='" + serviceInstanceOperation + '\'' +
                    '}';
        }
    }

    /**
     * 服务实例操作
     */
    class ServiceInstanceOperation {
        /**
         * 服务注册
         */
        public final static String REGISTER = "register";
        /**
         * 服务删除
         */
        public static final String REMOVE = "remove";
    }
}
